# -*- coding: utf-8 -*-
"""
项目名称：NoSql_kafka_project
文件名称：spark_hbase_redis.py
创建时间：2025-06-05 16:21:37

系统用户：Administrator
作　　者：無以菱
联系邮箱：huangjing2001.guet@qq.com
功能描述：
- 功能：从HBase读取数据，按支付方式统计金额，结果存入Redis
- 分析频率：每5秒执行一次
- 输出：JSON格式的统计结果
"""
from pyspark.sql import SparkSession
from pyspark.sql.functions import sum, count, avg
from pyspark.sql.types import StructType, StructField, StringType, FloatType
import happybase
import redis
import json
import traceback
import time
import sys
from config import REDIS_CONFIG

# 初始化Redis连接（使用配置文件参数）
def get_redis_client():
    try:
        client = redis.Redis(
            host=REDIS_CONFIG['host'],
            port=REDIS_CONFIG['port'],
            db=REDIS_CONFIG['db']
        )
        client.ping()
        print("Redis连接成功")
        return client
    except Exception as e:
        print(f"Redis连接失败: {e}")
        return None

redis_client = get_redis_client()

def run_spark_analysis():
    """
    从HBase读取订单数据，使用Spark进行分析，并将结果存入Redis
    
    分析内容：按支付方式统计支付金额总和
    """
    spark = None
    hbase_conn = None
    
    try:
        # 初始化Spark会话
        spark = SparkSession.builder \
            .appName("HBaseToRedis") \
            .master("local[*]") \
            .getOrCreate()
        
        print("Spark会话已创建")
        
        # 连接HBase
        hbase_conn = happybase.Connection('192.168.10.10', port=9090)
        table = hbase_conn.table('real_time_orders_table')
        
        print("HBase连接成功，开始读取数据")
        
        # 读取HBase数据
        rows = table.scan()
        data = []
        for row_key, cols in rows:
            try:
                # 解析每行数据
                data.append({
                    "PAYWAY": cols[b'cf:PAYWAY'].decode('utf-8'),
                    "PAY_AMOUNT": float(cols[b'cf:PAY_AMOUNT'])
                })
            except (KeyError, ValueError) as e:
                print(f"处理行 {row_key} 时出错: {e}")
                continue
        
        # 如果没有数据，提前返回
        if not data:
            print("没有找到数据，跳过分析")
            return
            
        # 定义Schema
        schema = StructType([
            StructField("PAYWAY", StringType(), True),
            StructField("PAY_AMOUNT", FloatType(), True)
        ])
        
        # 使用Spark创建DataFrame
        print("创建Spark DataFrame...")
        rdd = spark.sparkContext.parallelize(data)
        spark_df = spark.createDataFrame(rdd, schema)
        
        # 使用Spark进行数据分析
        print("使用Spark进行数据分析...")
        stats = spark_df.groupBy("PAYWAY").agg(
            sum("PAY_AMOUNT").alias("TOTAL_AMOUNT"),
            count("*").alias("ORDER_COUNT"),
            avg("PAY_AMOUNT").alias("AVG_AMOUNT")
        ).orderBy("TOTAL_AMOUNT", ascending=False)
        
        # 显示分析结果
        print("\n支付方式统计结果:")
        stats.show()
        
        # 将结果转换为JSON并存入Redis
        stats_json = stats.toJSON().collect()
        stats_list = [json.loads(row) for row in stats_json]
        stats_str = json.dumps(stats_list)
        
        if redis_client:
            redis_client.set(REDIS_CONFIG['stats_key'], stats_str)
            print(f"更新Redis统计结果: {stats_list}")
        else:
            print(f"Redis客户端未连接，无法保存结果: {stats_list}")
    
    except Exception as e:
        print(f"Spark分析过程中发生错误: {e}")
        traceback.print_exc()
    
    finally:
        # 关闭连接
        if spark:
            spark.stop()
        if hbase_conn:
            hbase_conn.close()
        print("Spark分析任务完成")

if __name__ == "__main__":
    # 测试
    print("正在启动Spark分析任务进行调试...")
    try:
        # 运行一次分析
        run_spark_analysis()
        
        # 或者循环运行，模拟在main.py中的行为
        run_continuously = input("是否持续运行分析任务? (y/n): ").lower() == 'y'
        if run_continuously:
            print("每5秒执行一次分析任务，按Ctrl+C停止...")
            while True:
                time.sleep(5)
                print("\n" + "-"*50)
                print(f"开始新一轮分析 - {time.strftime('%Y-%m-%d %H:%M:%S')}")
                run_spark_analysis()
    except KeyboardInterrupt:
        print("Spark分析任务已停止")
    except Exception as e:
        print(f"Spark分析任务运行出错: {e}")
        sys.exit(1)



